/*
 * Licensed to the Tai Ping Jin Ke
 *
 * Copyright (c) 2022 .
 * All rights reserved.
 * 项目名称：太平金科-基础组件封装层-SpringBoot组件封装-缺省实现
 * 版权说明：本软件属太平金融科技服务(上海)有限公司所有，在未获得太平金融科技服务(上海)有限公司正式授权情况下，任何企业和个人，不能获取、阅读、安装、传播本软件涉及的任何受知识产权保护的内容。
 */
package com.taipingframework.boot.web.websocket;

import com.taipingframework.utility.context.BeanLifecycle;
import com.taipingframework.utility.context.SpringContextHolder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;

import javax.websocket.OnClose;
import javax.websocket.OnError;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.Collection;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.LongAdder;

@Slf4j
@ServerEndpoint(value = "/websocket/{channelKey}")
public class WebSocketService extends BeanLifecycle {

    private static ApplicationContext applicationContext;

    @Override
    protected void initMethod() {
        applicationContext = SpringContextHolder.getApplicationContext();
    }

    /**
     * WebSocket会话计数器
     */
    private static LongAdder counter = new LongAdder();
    /**
     * channel-key与会话的绑定关系
     */
    private static ConcurrentHashMap<String, CopyOnWriteArrayList<Session>> channel2SessionMap = new ConcurrentHashMap<>();

    /**
     * 建立连接
     */
    @OnOpen
    public void onOpen(@PathParam("channelKey") String channelKey, Session session) {
        counter.increment();
        log.debug("WebSocket-onOpen # 会话：{} 通道：{} 当前会话数量：{}", session.getId(), channelKey, counter.longValue());
        if (channel2SessionMap.containsKey(channelKey)) {
            channel2SessionMap.get(channelKey).add(session);
        } else {
            CopyOnWriteArrayList<Session> sessionList = new CopyOnWriteArrayList<>();
            sessionList.add(session);
            channel2SessionMap.put(channelKey, sessionList);
        }
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(@PathParam("channelKey") String channelKey, Session session) {
        counter.decrement();
        log.debug("WebSocket-onClose # 会话：{} 通道：{} 剩余会话数量：{}", session.getId(), channelKey, counter.longValue());
        if (channel2SessionMap.containsKey(channelKey)) {
            channel2SessionMap.get(channelKey).remove(session);
        }
    }

    /**
     * 发生错误
     */
    @OnError
    public void onError(@PathParam("channelKey") String channelKey, Session session, Throwable error) {
        log.error("WebSocket-onError # 会话：{} 通道：{} 发生异常", session.getId(), channelKey, error);
    }

    /**
     * 收到会话通道的消息
     *
     * @param message 消息
     * @param session 会话
     */
    @OnMessage
    public void onMessage(@PathParam("channelKey") String channelKey, Session session, String message) {
        log.debug("WebSocket-onMessage # 会话：{} 通道：{} 消息：{}", session.getId(), channelKey, message);

        if (Objects.nonNull(applicationContext)) {
            applicationContext.publishEvent(new WebSocketEvent(this, channelKey, session, message));
        }
    }

    /**
     * 向指定会话通道发送消息
     *
     * @param channelKey 指定通道唯一标识
     * @param message    消息
     */
    public void sendToClient(String channelKey, String message) {
        // channel2SessionMap.get(channelKey).getBasicRemote().sendText(message);
        if (channel2SessionMap.containsKey(channelKey)) {
            channel2SessionMap.get(channelKey)
                    .stream()
                    .parallel()
                    .forEach(session -> session.getAsyncRemote().sendText(message));
        }
    }

    /**
     * 向所有会话通道发送消息
     *
     * @param message 消息
     */
    public void sendToAllClient(String message) {
        channel2SessionMap.values()
                .stream()
                .flatMap(Collection::stream)
                .parallel()
                // session.getBasicRemote().sendText(message);
                .forEach(session -> session.getAsyncRemote().sendText(message));
    }

}
